package com.tmsb.fair_shop;

import com.alibaba.fastjson.JSON;
import com.tmsb.pojo.FairShopOrderMall;
import com.tmsb.pojo.FairShopOrderNetwork;
import com.tmsb.pojo.FairShopOrderVerifiedNetwork;
import com.tmsb.pojo.MysqlBinlog;
import com.tmsb.sink.OrderMallSink;
import com.tmsb.sink.OrderNetworkSink;
import com.tmsb.sink.OrderVerifiedNetworkSink;
import com.tmsb.utils.*;
import com.tmsb.utils.trigger.TimerTrigger;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import redis.clients.jedis.JedisCluster;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * Description:
 * Version:1.0.0
 *
 * @author wxw
 * @date 2020/7/24
 */
public class fairShopAnalysis {

    //TODO 每个指标的侧输出流标签
    private static final OutputTag<FairShopOrderNetwork> ORDER_NETWORK = new OutputTag<FairShopOrderNetwork>("order_network") {};
    private static final OutputTag<FairShopOrderVerifiedNetwork> ORDER_VERIFIED_NETWORK = new OutputTag<FairShopOrderVerifiedNetwork>("order_verified_network") {};
    private static final OutputTag<FairShopOrderMall> ORDER_MALL = new OutputTag<FairShopOrderMall>("order_mall") {};

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setStateBackend(new FsStateBackend("file:///C:\\Users\\wuxiangwu\\Desktop\\mall"));
        //TODO 设置checkpoint检查点
        env.setStateBackend(new FsStateBackend(ConfUtil.getFairShopAnalysisCheckpointUri()));

        //TODO 取消作业时保留检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.enableCheckpointing(ConfUtil.getCheckpointInterval(), CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(ConfUtil.getCheckpointMinPauseBetween());
        env.setParallelism(ConfUtil.getParallelism());

        //TODO 重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(
                ConfUtil.getfailureRate(), org.apache.flink.api.common.time.Time.of(ConfUtil.getfailureInterval(), TimeUnit.MINUTES),
                org.apache.flink.api.common.time.Time.of(ConfUtil.getdelayInterval(), TimeUnit.SECONDS)
        ));


        //TODO source
        DataStream<String> source = env
                .addSource(new FlinkKafkaConsumer<>(ConfUtil.getTopicOfFairShop(), new SimpleStringSchema(), ConfUtil.getKafkaProperties()))
                .uid("kafka-source-fs");

        //TODO 根据需要的指标进行分流操作
        SingleOutputStreamOperator<Object> process = source
                .map(s -> JSON.parseObject(s, MysqlBinlog.class)).uid("map-parser-hm")
                .process(new ProcessFunction<MysqlBinlog, Object>() {
                    @Override
                    public void processElement(MysqlBinlog mysqlBinlog, Context context, Collector<Object> collector) throws Exception {
                        //TODO 网点每日进货量指标侧输出流
                        if (Const.MysqlDatabase.TMSB_FAIR_SHOP.toString().equalsIgnoreCase(mysqlBinlog.getDatabase())) {
                            if (Const.MysqlTable.ORDER_NETWORK.toString().equalsIgnoreCase(mysqlBinlog.getTable())) {
                                FairShopOrderNetwork fairShopOrderNetwork = JSON.parseObject(mysqlBinlog.getData(), FairShopOrderNetwork.class);
                                if (fairShopOrderNetwork == null) {
                                    return;
                                }
                                context.output(ORDER_NETWORK, fairShopOrderNetwork);
                               //TODO 网点每日核销水数量指标侧输出流
                            } else if (Const.MysqlTable.ORDER_VERIFIED_NETWORK.toString().equalsIgnoreCase(mysqlBinlog.getTable())) {
                                FairShopOrderVerifiedNetwork fairShopOrderVerifiedNetwork = JSON.parseObject(mysqlBinlog.getData(), FairShopOrderVerifiedNetwork.class);
                                if (fairShopOrderVerifiedNetwork == null) {
                                    return;
                                }
                                context.output(ORDER_VERIFIED_NETWORK, fairShopOrderVerifiedNetwork);
                            //TODO 泰木山泉每日订单指标侧输出流
                            } else if (Const.MysqlTable.ORDER_MALL.toString().equalsIgnoreCase(mysqlBinlog.getTable())) {
                                FairShopOrderMall fairShopOrderMall = JSON.parseObject(mysqlBinlog.getData(), FairShopOrderMall.class);
                                if (fairShopOrderMall == null) {
                                    return;
                                }
                                context.output(ORDER_MALL, fairShopOrderMall);
                            }
                        }
                    }
                }).uid("process-side-output-fs");

        //TODO 网点每日进货量
        process.getSideOutput(ORDER_NETWORK)
                .filter(o -> {
                    long order_time = o.getOrder_time() - 28800000;
                    Long zeroPointTimestamps = Util.getZeroPointTimestamps(order_time);
                    return (o.getOrder_status() == 1 && order_time > zeroPointTimestamps) || (o.getOrder_status() == -3 && order_time > zeroPointTimestamps)||(o.getOrder_status() == -1 && order_time > zeroPointTimestamps);
                }).uid("filter")
               .addSink(new OrderNetworkSink())
                 .uid("network-sink");
     /*   //开窗，每日滚动
        filterStream.windowAll(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
               .trigger(new TimerTrigger<FairShopOrderNetwork>())
                .process(new FairShopOrderNetworkProcess())
                .uid("FairShopOrderNetworkProcess");
*/
        //TODO 网点每日核销水数量
        process.getSideOutput(ORDER_VERIFIED_NETWORK)
                .filter(o -> {
                    long create_time = o.getCreate_time() - 28800000;
                    Long zeroPointTimestamps = Util.getZeroPointTimestamps(create_time);
                    return (o.getIs_deleted() == 0 && create_time > zeroPointTimestamps) ;
                }).uid("filter_verify")
                .addSink(new OrderVerifiedNetworkSink())
                .uid("verified-sink");

     //TODO  泰木山泉每日订单
        process.getSideOutput(ORDER_MALL)
                .filter(o -> {
                    long order_time = o.getOrder_time() - 28800000;
                    Long zeroPointTimestamps = Util.getZeroPointTimestamps(order_time);
                    return (o.getStatus() == 1 && order_time > zeroPointTimestamps) || (o.getStatus() == -3 && order_time > zeroPointTimestamps);
                }
              ).uid("filter_order_mall")
                .addSink(new OrderMallSink())
                .uid("orderMall-sink");

        try {
            env.execute("fairShopAnalysis");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

/*    private static class FairShopOrderNetworkProcess extends ProcessAllWindowFunction<FairShopOrderNetwork,String, TimeWindow>{
        //TODO 总数量
        private ValueState<String> total_quantity;
        //TODO 订单id集合,用于判断是否重复算过
        private ValueState<Set> idState;
        //TODO 当前窗口最大时间戳
        private ValueState<Long> windowEnd;
        private String date;
        private SimpleDateFormat sdf;
        private JedisCluster jedis;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
           idState = getRuntimeContext().getState(new ValueStateDescriptor<Set>("idState",Set.class,new HashSet<>()));
           total_quantity = getRuntimeContext().getState(new ValueStateDescriptor<String>("total_quantity",String.class,"0"));
            windowEnd = getRuntimeContext().getState(new ValueStateDescriptor<Long>("windowEnd",Long.class,0L));
            sdf=new SimpleDateFormat("yyyyMMdd");
            jedis = RedisCluster.getInstance(ConfUtil.getRedisCluster()).getRedisClusterList().get(0);
            System.out.println("进入open方法");
        }



        @Override
        public void process(Context context, Iterable<FairShopOrderNetwork> iterable, Collector<String> collector) throws Exception {
            System.out.println("进入process 方法");
            long end = context.window().maxTimestamp();
            if(date == null) {
                date = sdf.format(new Date(end));
            }
            //TODO 新的窗口就清空状态，并更新windowEnd的状态
            if (windowEnd.value()!=end){
                windowEnd.update(end);
                idState.clear();
                total_quantity.clear();
                date = sdf.format(new Date(end));
            }
             Set idSet = idState.value();
             String sum = total_quantity.value();
             Iterator<FairShopOrderNetwork> iterator = iterable.iterator();
             while (iterator.hasNext()){
                  FairShopOrderNetwork on = iterator.next();
                  long id = on.getId();
                 //TODO 已付款，待配送
                 long quantity;
                  if(on.getOrder_status()==1){
                     //TODO 如果订单状态为1的已经来过一次，不要重复计算
                      if(idSet.contains(id)){
                          continue;
                      }
                      //TODO 新增订单
                      quantity = on.getQuantity();
                      idSet.add(id);
                  }else {
                      //TODO 订单状态为3，即退款
                      if(!idSet.contains(id)){
                          continue;
                      }
                      quantity = on.getQuantity()*(-1);
                      idSet.remove(id);
                  }
                 sum += quantity;
                  //TODO 向redis发送数据
                 jedis.hset(ConfUtil.getOrderNetworkQuantityRedisKey(),date,sum+"");
                 System.out.println("quantity总和为："+sum);
             }
             //TODO 更新状态
            total_quantity.update(sum);
             idState.update(idSet);


        }

        @Override
        public void close() throws Exception {
            super.close();
            jedis.close();
        }
    }*/


}
